package com.ehualu.liaocheng

import java.text.SimpleDateFormat

import org.apache.spark.rdd.RDD
import org.apache.spark.sql.SparkSession

import scala.collection.mutable.ArrayBuffer

/**
  * @: 吴敬超
  * @: 2019/8/29 17:33
  */
object lxydzs {

  def main(args: Array[String]): Unit = {

    val spark = SparkSession.builder()
      .master("local[4]")
      .appName("BanSui")
      .getOrCreate()

    // 获取SparkContext实例对象
    val sc = spark.sparkContext
    // 设置日志级别
    //    sc.setLogLevel("WARN")


    val dataRDD: RDD[String] = spark.sparkContext.textFile("E:\\0830\\gcxx").distinct()


    //    e表示每一行
    val transRDD = dataRDD.map(e => {
      val arr: Array[String] = e.split(",")
      //      时间
      val sj = arr(3).substring(0, 19)

      //      号牌
      val hphm = arr(5)

      //      卡口编号
      val kkbh = arr(1)

      //      返回三元组
      (hphm, kkbh, sj)

      //      处理map结果

      //      filter 过滤  _._1   是hphm
    }).filter(!_._1.toString.equals("车牌"))

    val count1: Long = transRDD.count()
    //    println(count1)   //18948

    //    val hphmGroupedRDD: RDD[(String, Iterable[String])] = transRDD.map(e => {

    val hphmGroupedRDD = transRDD.map(e => {
      //      var A: Map[String, String] = Map()
      //以车牌号为key  其他的放在元组里
      val t = (e._2, e._3)
      //      xxxx ,
      (e._1, t)
    }).groupByKey()
    hphmGroupedRDD.take(100).foreach(println)

    hphmGroupedRDD.count();
    val filterHphmGroupedRDD = hphmGroupedRDD.filter(e => {
      e._2.size > 1
    })
//    filterHphmGroupedRDD.take(100).foreach(println)

    val count3: Long = filterHphmGroupedRDD.count()
    println(s"count: $count3") //3150
    //    println("7777777777777777777777777777777777777777777")

    val roadTimeRDD = filterHphmGroupedRDD.map(iter => {
      //      println("7777777777777777777777777777777777777777777")
      //      println(iter._1)


      var array = ArrayBuffer[(String, String)]()
      val iterator = iter._2.iterator
      //取出每个key 的对应的信息
      while (iterator.hasNext) {

        //        Iterator类的next( )方法在同一循环中不能出现两次。
        //        println(iterator.next())
        array.append(iterator.next())

      }

      val sortedTupleArray: ArrayBuffer[(String, String)] = array.sortBy(_._2)

      //      println(sortedTupleArray)
      val size: Int = sortedTupleArray.size
      var rearray = ArrayBuffer[(String, Long)]()
      for (i <- (0 to size - 2)) {

        var kkbh1 = sortedTupleArray(i)._1
        var kkbh2 = sortedTupleArray(i + 1)._1
        var kksj1 = sortedTupleArray(i)._2
        var kksj2 = sortedTupleArray(i + 1)._2

        var ldbh = kkbh1 + "_" + kkbh2
        var ldsj = getDifferentSeconds(kksj1, kksj2)


//        12345_23456   500
        var ldyz = (ldbh, ldsj)
        rearray.append(ldyz)
      }
      rearray

      //      for(i <- 0 until sortedTupleArray.length){
      //        println(sortedTupleArray(i))
      //
      //
      //
      //      }
      //      val diffenenceSeconds = getDifferentSeconds(sj1, sj2)
      //      println(iter._2)

    }

    )
    //    roadTimeRDD.take(100).foreach(println)

    roadTimeRDD.count()

    val notNullRoadRDD = roadTimeRDD.filter(e => {
      e.size > 0
    })

    //    notNullRoadRDD.map((_, 1)).groupBy(_._1)
    println("8888888888888888888888888888888888888")

    //    notNullRoadRDD.foreach(println)
    //
    //    notNullRoadRDD.count()

    //
    //    (37150002324_37150002323,2)
    //    (37150002324_37150002323,22)
    //    (37150002324_37150002323,4)
    //    (37150002324_37150002323,3)
    //    (37150002324_37150002323,9)
    //    (3715000916_3715000916,15491)
    //    (3715000916_3715000916,2943)
    //    (3715000916_3715000916,2897)
    //    (3715000916_3715000916,11628)
    //    (3715000916_3715000916,6858)
    //    (3715000916_37150002324,41518)
    //


    //    notNullRoadRDD.flatMap(x=>x).foreach(println(_))

    //
    //    (37150002322_37150002322,CompactBuffer(167, 25229, 33481, 32468, 25178, 36636, 13627, 20488, 19096, 7057, 4669, 19367, 20098, 35439, 14731, 24269, 29231, 5950, 25766, 14665, 24191, 22960, 34038, 30073, 17, 41892, 16020, 6184, 16396, 28850, 9172, 2292, 24253, 5772, 18928, 59418, 22210, 24608, 45960, 14303, 22250, 19074, 12824, 16838, 19949, 15463, 4391, 17861, 11023, 21390, 26700, 5527, 46507, 43468, 27961, 22408, 12614, 11622, 23470, 38952, 19334, 22571, 41598, 13169, 31160, 4173, 7389, 67684, 35974, 23482, 22431, 10499, 35127, 17116, 26324, 17978))
    //    (3715000916_37150002323,CompactBuffer(30728, 9082, 33018, 24168, 4956, 6550, 37948, 6846, 6018))
    //    (37150002323_37150002323,CompactBuffer(2081, 13736, 22435, 18363, 38447, 21672, 9596, 32641, 32178, 28, 3869, 77995, 39852, 13141, 8219, 10270, 24335, 10199, 6829, 13247, 14133, 3588, 8333, 41620, 26961, 28715, 4251))
    //notNullRoadRDD.flatMap(x => x).groupByKey()
    val luDuanRdd = notNullRoadRDD.flatMap(x => x).groupByKey()

    val luduantjrdd = luDuanRdd.map(

      line => {
        println("&&&&&&&&路段编号&&&&&&&&&" + line._1)
        //        line._1
        var sjnum = line._2.iterator.size

        println("*******sjnum******" + sjnum)

        val sjiterator = line._2.iterator
        var sumsj: Long = 0
        //取出每个key 的对应的信息
        while (sjiterator.hasNext) {
          sumsj = sumsj + sjiterator.next()
          //        Iterator类的next( )方法在同一循环中不能出现两次。
          //        println(iterator.next())

        }
        var ldpjsj = sumsj / sjnum

        println("********路段平均时间***************：" + ldpjsj)

        (line._1, ldpjsj)
      }
    )

    //    luduantjrdd.foreach(println(_))
    spark.close()
  }

  def getDifferentSeconds(start: String, end: String): Long = {

    val fm = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss")
    val startLong = fm.parse(start).getTime
    val endLong = fm.parse(end).getTime

    val seconds = (endLong - startLong) / 1000
    seconds
  }

}
